[FLINK-20625][pubsub,e2e] Add PubSubSource connector using FLIP-27 #2

RyanSkraba · 2023-01-26T01:18:24Z

What is the purpose of the change

Originally from PR [FLINK-20625][pubsub,e2e] Add Google Cloud Pub/Sub source connector in accordance with FLIP-27 flink#15152, then [FLINK-20625][pubsub,e2e] Add PubSubSource connector using FLIP-27 flink#18823 to help finish the PR for a new Pub/Sub connector that implements the Source API from FLIP-27

Brief change log

Rebased [FLINK-20625][pubsub,e2e] Add Google Cloud Pub/Sub source connector in accordance with FLIP-27 flink#15152 on master and fixed conflicts
- Squashed the original changes (retaining @jakob-ed's authorship)
- Moved new classes to packages that align with the FLIP-27 KafkaSource
Modified the E2E tests to fetch a bounded number of records from the Pub/Sub emulator
Rebased [FLINK-20625][pubsub,e2e] Add PubSubSource connector using FLIP-27 flink#18823 on externalized repository main, re-homing the change to the new directory structure.

Verifying this change

Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing

This change added tests and can be verified as follows:

Adds end-to-end test using the Pub/Sub emulator.

(example:)

Added integration tests for end-to-end deployment with large payloads (100MB)
Extended integration test for recovery after master (JobManager) failure
Added test that validates that TaskInfo is transferred only once across recoveries
Manually verified the change by running a 4 node cluser with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? docs / JavaDocs

boring-cyborg · 2023-01-26T01:18:28Z

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

* WIP * WIP * Working WIP * Clean up * Place new Pub/Sub source into existing Pub/Sub connector module * Clean up * Apply Spotless code formatting [FLINK-20625][pubsub,e2e] Attempt to support stopping the reader when stopmark is encountered [FLINK-20625][pubsub,e2e] Add checkpointing and do some refactorings * Simplify fetching from Pub/Sub in SplitReader * Allow Pub/Sub source to be only continuous unbounded * Add basic PubSubSource builder * Add configuration options for SubscriberFactory to PubSubSource, remove unused collector * Add checkpointing [FLINK-20625][pubsub,e2e] Allow multiple records inside single Pub/Sub message for deserialization [FLINK-20625][pubsub,e2e] Add Javadocs, README and clean up [FLINK-20625][pubsub,e2e] Reduce visibility of classes and their members [FLINK-20625][pubsub,e2e] Propagate Pub/Sub subscriber creation errors from SplitReader [FLINK-20625][pubsub,e2e] Use constants for default Pub/Sub subscriber parameters [FLINK-20625][pubsub,e2e] Fix dynamic Scala version in artifact example [FLINK-20625][pubsub,e2e] Rename PubSubEnumeratorCheckpoint -> PubSubEnumeratorState [FLINK-20625][pubsub,e2e] Add version checks for deserialization [FLINK-20625][pubsub,e2e] Remove unnecessary declaration of exception-throwing [FLINK-20625][pubsub,e2e] Remove disfunctional end-of-stream logic [FLINK-20625][pubsub,e2e] Avoid concurrency issues with list of Pub/Sub messages to acknowledge [FLINK-20625][pubsub,e2e] Refactor PubSubSourceBuilder [FLINK-20625][pubsub,e2e] Clarify consistency guarantee description [FLINK-20625][pubsub,e2e] Clarify Pub/Sub request timeout [FLINK-20625][pubsub,e2e] Restructure and extend readme, add basic architecture info to docstring [FLINK-20625][pubsub,e2e] Attempt to solve concurrency issues with checkpointing

MartijnVisser · 2023-06-28T07:35:36Z

@RyanSkraba Do you have any idea who could help with a review here?

XComp · 2023-10-04T10:28:33Z

@dchristle how about you? 😉 Would you have time to look at this?

...ctor-gcp-pubsub/src/main/java/org/apache/flink/connector/gcp/pubsub/source/PubSubSource.java

...sub/src/main/java/org/apache/flink/connector/gcp/pubsub/source/reader/PubSubSplitReader.java

dchristle · 2023-11-10T23:48:26Z

Hi @RyanSkraba,

Thank you for shepherding and improving this connector implementation over the years. I'm hoping to help review & improve it so we can finally get it merged. I just opened RyanSkraba#1 to add some improvements based on my experience running this code in production. Our internal implementation has some other improvements, but they need to be properly cleaned up & can come in a later PR, as they aren't essential. Please take a look when you get a chance.

* Add retry mechanism to acknowledging pubsub messages * Remove synchronized locks and instead decouple reader and fetcher threads with ArrayBlockingQueue. This fixes a concurrency bug between separate threads calling fetch/acknowledgeMessages. * Handle failed checkpoints and ensure subscriber is not null in acknowledgeMessages. * Implement splitAckIds with iteration instead of recursion. Fixes bug where payload can exceed maxPayload. --------- Co-authored-by: Richard Deurwaarder <[email protected]>

RyanSkraba · 2023-11-15T14:45:29Z

Just a quick update on this PR! Thanks so much @dchristle for the review and the improvements -- I've included everything here in the PR and added you as a collaborator to this branch. Please feel free to apply any additional changes that you've learned from your use in production! That's really the most valuable source of info for a new connector.

I've applied the changes you've requested, as well as some other very minor code clean-ups suggested through the IDE (some unused exceptions, another missing Java generic and using a deprecated super-constructor).

snuyanzin · 2024-04-15T08:08:32Z

@dchristle thanks for the review
May I ask you to have another iteration review for this PR?

vahmed-hamdy

I have a general question, if we are delaying acks to checkpoints, how are we going to handle if checkpoint duration is bigger than ack deadline on pubsub source, also we are having multiple pull requests ahead of acking, this causes multiple duplicates IIUC. This is not tested in IT tests added.

vahmed-hamdy · 2024-06-05T09:26:34Z

...ctor-gcp-pubsub/src/main/java/org/apache/flink/connector/gcp/pubsub/source/PubSubSource.java

+        implements Source<OUT, PubSubSplit, PubSubEnumeratorState>, ResultTypeQueryable<OUT> {
+    private final PubSubDeserializationSchema<OUT> deserializationSchema;
+    private final PubSubSubscriberFactory pubSubSubscriberFactory;
+    private final Properties props;


nit: rename props, is t client props, reader props,....?

vahmed-hamdy · 2024-06-05T09:33:53Z

...ctor-gcp-pubsub/src/main/java/org/apache/flink/connector/gcp/pubsub/source/PubSubSource.java

+        private String subscriptionName;
+        private PubSubSubscriberFactory pubSubSubscriberFactory;
+        private Properties props;
+        private Credentials credentials;


we should pass credential provider instead, to natively support dynamic credentials discovery and refreshing.

vahmed-hamdy · 2024-06-05T09:45:47Z

...ctor-gcp-pubsub/src/main/java/org/apache/flink/connector/gcp/pubsub/source/PubSubSource.java

+                        new DefaultPubSubSubscriberFactory(
+                                ProjectSubscriptionName.format(projectName, subscriptionName),
+                                DEFAULT_PUBSUB_SUBSCRIBER_NUMBER_OF_RETRIES,
+                                Duration.ofSeconds(15),


nit: why not expose as default like the others?

vahmed-hamdy · 2024-06-05T09:49:51Z

...ctor-gcp-pubsub/src/main/java/org/apache/flink/connector/gcp/pubsub/source/PubSubSource.java

+ *
+ * @param <OUT> The output type of the source.
+ */
+public class PubSubSource<OUT>


missing tag, @PublicEvolving or @Public

vahmed-hamdy · 2024-06-05T10:03:32Z

...ain/java/org/apache/flink/connector/gcp/pubsub/source/enumerator/PubSubSourceEnumerator.java

+ * The enumerator for the {@link PubSubSource}. It does not do any work discovery as envisioned by
+ * FLIP-27 because GCP Pub/Sub hides partitions and other implementation details.
+ */
+public class PubSubSourceEnumerator implements SplitEnumerator<PubSubSplit, PubSubEnumeratorState> {


missing annotation, @Internal

vahmed-hamdy · 2024-06-06T11:08:13Z

...ain/java/org/apache/flink/connector/gcp/pubsub/source/reader/PubSubSourceFetcherManager.java

+            FutureCompletingBlockingQueue<RecordsWithSplitIds<Tuple2<T, Long>>> elementsQueue,
+            Supplier<SplitReader<Tuple2<T, Long>, PubSubSplit>> splitReaderSupplier,
+            Configuration config) {
+        super(elementsQueue, splitReaderSupplier, config);


Deprecated constructor

vahmed-hamdy · 2024-06-06T11:22:45Z

...ain/java/org/apache/flink/connector/gcp/pubsub/source/reader/PubSubSourceFetcherManager.java

+        }
+    }
+
+    private void enqueuePrepareForAcknowledgementTask(


We are calling this on onCheckpointComplete where we enqueue a task that would possibly not start immediately. What happens if we fail during acking.
If we move the acking responsiblity to Source reader, that would be much cleaner and we could make it synchronous

vahmed-hamdy · 2024-06-06T11:31:54Z

...tests/src/test/java/org/apache/flink/connector/gcp/pubsub/EmulatedPubSubSourceTest.java.orig

@@ -0,0 +1,181 @@
+/*


The file extension is not correct, also Maybe we rename it EmulatedPubSubSourceV2Test

vahmed-hamdy · 2024-06-06T11:32:49Z

...-e2e-tests/src/test/java/org/apache/flink/connector/gcp/pubsub/EmulatedPubSubSourceTest.java

+ *
+ */
+
+package org.apache.flink.connector.gcp.pubsub;


It should be enough to add to e2e tests module

vahmed-hamdy · 2024-06-06T11:34:33Z

...-e2e-tests/src/test/java/org/apache/flink/connector/gcp/pubsub/EmulatedPubSubSourceTest.java

+import static org.junit.Assert.assertTrue;
+
+/** Test of {@link PubSubSource} against the GCP Pub/Sub emulator SDK. */
+public class EmulatedPubSubSourceTest extends GCloudUnitTestBase {


We should use Junit5

markma0215 · 2024-09-18T17:11:32Z

could I know when this PR will be merged and release in which version? thx

boring-cyborg bot added the component=Connectors/Google Cloud PubSub label Jan 26, 2023

RyanSkraba mentioned this pull request Jan 26, 2023

[FLINK-20625][pubsub,e2e] Add PubSubSource connector using FLIP-27 apache/flink#18823

Closed

MartijnVisser force-pushed the main branch from 3607c7a to d32b1f9 Compare March 13, 2023 14:13

JakobEdding and others added 5 commits June 28, 2023 09:07

[FLINK-20625][pubsub,e2e] Update parameter for EnumeratorState

81775b0

[FLINK-20625][pubsub] Align new source with Kafka package names

e81bac4

[FLINK-20625] Fix end-to-end test stopping condition

459d773

[docs] Minor documentation fixes

3b33d92

MartijnVisser force-pushed the rskraba/FLINK-20625-pubsub-PR18823 branch from 7813345 to 3b33d92 Compare June 28, 2023 07:10

dchristle mentioned this pull request Nov 10, 2023

Improvements to FLINK-20625 FLIP-27 connector RyanSkraba/flink-connector-gcp-pubsub#1

Merged

dchristle reviewed Nov 10, 2023

View reviewed changes

...ctor-gcp-pubsub/src/main/java/org/apache/flink/connector/gcp/pubsub/source/PubSubSource.java Outdated Show resolved Hide resolved

...sub/src/main/java/org/apache/flink/connector/gcp/pubsub/source/reader/PubSubSplitReader.java Outdated Show resolved Hide resolved

dchristle and others added 3 commits November 13, 2023 13:49

[hotfix] Spotless code

55f0cf1

[FLINK-20625] Review comments and code clean-up

b6e38a6

RyanSkraba mentioned this pull request Nov 15, 2023

[FLINK-13230] [pubsub] Add retry mechanism to acknowledging pubsub me… apache/flink#9388

Closed

vahmed-hamdy suggested changes Jun 6, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-20625][pubsub,e2e] Add PubSubSource connector using FLIP-27 #2

[FLINK-20625][pubsub,e2e] Add PubSubSource connector using FLIP-27 #2

RyanSkraba commented Jan 26, 2023

boring-cyborg bot commented Jan 26, 2023

MartijnVisser commented Jun 28, 2023

XComp commented Oct 4, 2023 •

edited

Loading

dchristle commented Nov 10, 2023

RyanSkraba commented Nov 15, 2023

snuyanzin commented Apr 15, 2024

vahmed-hamdy left a comment •

edited

Loading

vahmed-hamdy Jun 5, 2024

vahmed-hamdy Jun 5, 2024

vahmed-hamdy Jun 5, 2024

vahmed-hamdy Jun 5, 2024

vahmed-hamdy Jun 5, 2024

vahmed-hamdy Jun 6, 2024

vahmed-hamdy Jun 6, 2024

vahmed-hamdy Jun 6, 2024

vahmed-hamdy Jun 6, 2024

vahmed-hamdy Jun 6, 2024

markma0215 commented Sep 18, 2024

[FLINK-20625][pubsub,e2e] Add PubSubSource connector using FLIP-27 #2

Are you sure you want to change the base?

[FLINK-20625][pubsub,e2e] Add PubSubSource connector using FLIP-27 #2

Conversation

RyanSkraba commented Jan 26, 2023

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

boring-cyborg bot commented Jan 26, 2023

MartijnVisser commented Jun 28, 2023

XComp commented Oct 4, 2023 • edited Loading

dchristle commented Nov 10, 2023

RyanSkraba commented Nov 15, 2023

snuyanzin commented Apr 15, 2024

vahmed-hamdy left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

markma0215 commented Sep 18, 2024

XComp commented Oct 4, 2023 •

edited

Loading

vahmed-hamdy left a comment •

edited

Loading